package org.example.checkpoint;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class SavePointDemo {
    public static void main(String[] args) throws Exception {

        Configuration configuration = new Configuration();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

        //存储hdfs，指定hdfs依赖，指定访问hdfs的用户名
        // System.setProperty("HADOOP_USER_NAME", "root");

        //1、启用检查点：默认是Barrier对齐，周期5S，精准一次。
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //2、指定检查点的存储位置
        checkpointConfig.setCheckpointStorage("file:///opt/module/flink-1.17.2/checkpoint");
        //3、指定检查点的超时时间，默认为10分钟
        checkpointConfig.setCheckpointTimeout(60000);
        //4、设置检查点的最大并发，同时运行的checkpoint的最大数量
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        //5、最小等待间隔，上一轮的checkpoint结束到下一轮开始的最小等待时间,设置>0,并发就会变为1。
        checkpointConfig.setMinPauseBetweenCheckpoints(1000);
        //6、取消作业时checkpoint的数据，是否保留在外部系统，生产系统中建议保留
        //DELETE_ON_CANCELLATION：主动Cancel时，删除存在外部系统的chk-xx目录，如果程序突然挂掉不会删除。
        //RETAIN_ON_CANCELLATION：主动Cancel时，保留存在外部系统的chk-xx目录。
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //7、设置作业连续失败的次数，默认值为0，表示checkpoint一失败JOB的挂掉。
        checkpointConfig.setTolerableCheckpointFailureNumber(10);

        //开启非对齐的检查点(Barrier非对齐)
        //开启后，checkpoint模式必需是精准一次，最大并发数只能为1.
        checkpointConfig.enableUnalignedCheckpoints();
        //开启非对齐的checkpoint时才生效，默认为0，表示一开始就用非对齐的。
        //如果>0，一开始用对齐的检查点，超过超时时间，自动切换为非对齐的检查点。
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));


        env.socketTextStream("192.168.99.1", 9999)
                .uid("socket")
                .flatMap((FlatMapFunction<String, String>) (value, out) -> {
                    String[] split = value.split(" ");
                    for (String s : split) {
                        out.collect(s);
                    }
                })
                .uid("flatMap-wc")
                .name("wc-flatMap-name")
                .returns(Types.STRING)
                .map(value -> Tuple2.of(value, 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT)) //解析Lambda类型擦除的问题。
                .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
                .sum(1)
                .uid("sum-wc")
                .print()
                .uid("print-wc");
        env.execute("WordCountStreamUnbound");
    }
}
